package hadoop.hdfs;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Files;

/**
 * 关于配置的优先级：代码中最高>资源库目录下的>服务器上的>默认的
 */
public class HdfsProcess {
    FileSystem fileSystem = null;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        String HDFS_URI = "hdfs://hdphost:9000/";
        String USER = "root";
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("dfs.client.use.datanode.hostname", "true");
        conf.set("dfs.replication", "1");
        fileSystem = FileSystem.get(new URI(HDFS_URI), conf, USER);
    }

    @After
    public void destory() throws Exception {
        fileSystem.close();
    }

    /**
     * 根据指定的hdfs路径创建hdfs目录
     */
    @Test
    public void makeDir() throws Exception {
        Path hdfsDir = new Path("/output");
        if (!fileSystem.exists(hdfsDir)) {
            boolean isok = fileSystem.mkdirs(hdfsDir);
            if (isok) {
                System.out.println("目录创建成功");
            } else {
                System.out.println("目录创建失败");
            }
        } else {
            System.out.println("目录已经存在");
        }
    }

    /**
     * 上传文件
     */
    @Test
    public void uploadFile() throws Exception {
        Path srcPath = new Path("C:\\Users\\Administrator\\Desktop\\hadoop.txt");   //原目录
        Path dstPath = new Path("/output");  //目标目录
        fileSystem.copyFromLocalFile(false, srcPath, dstPath);
        FileStatus[] fileStatuses = fileSystem.listStatus(dstPath);
        for (FileStatus file : fileStatuses) {
            System.out.println(file.getPath());
        }
    }

    /**
     * 删除文件
     */
    @Test
    public void deleteFile() throws Exception {
        Path dstPath = new Path("/output");  //目标目录
        boolean isok = fileSystem.deleteOnExit(dstPath);
        if (isok) {
            System.out.println("删除成功");
        } else {
            System.out.println("删除失败");
        }
    }


    /**
     * 下载文件
     */
    @Test
    public void download() throws Exception {
        Path filePath = new Path("/user/root/hadoop.txt");
        try (InputStream is = fileSystem.open(filePath);
             OutputStream os = Files.newOutputStream(new File("D://a.txt").toPath());) {
            IOUtils.copy(is, os, 1024);
        }

        //方法2
        fileSystem.copyToLocalFile(true, filePath, new Path("D://b.txt"));
    }

    /**
     * 按照块下载文件
     */
    @Test
    public void downLoadSeek() throws Exception {
        Path filePath = new Path("/user/root/hadoop.txt");
        try(FSDataInputStream is = fileSystem.open(filePath);
            OutputStream os = Files.newOutputStream(new File("D://a.txt.part1").toPath())){
            is.seek(1024 * 1024 * 128);  //128M  从文件的128开始
            IOUtils.copy(is, os, 1024);
        }
    }


    /**
     * 文件以及目录信息查看
     */
    @Test
    public void fileInfo() throws Exception {
        Path filePath = new Path("/");
        RemoteIterator<LocatedFileStatus> listFile = fileSystem.listFiles(filePath, true);
        while (listFile.hasNext()) {
            LocatedFileStatus fileStatus = listFile.next();
            System.out.println("文件名：" + fileStatus.getPath().getName());
            System.out.println("文件权限：" + fileStatus.getPermission());
            System.out.println("文件长度：" + fileStatus.getLen());

            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation b : blockLocations) {
                String[] hosts = b.getHosts();
                for (String host : hosts) {
                    System.out.println("块" + host);
                }
            }
            System.out.println("----------");
        }
    }

}
